package com.inspinia.base.mq;

import com.google.common.collect.Lists;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Clock;

/**
 * 消费者死循环线程
 * Created by veblen on 16/5/3.
 */
public class ConsumerLoop implements Runnable {

    private final String topic;
    private final KafkaConsumer<String, String> consumer;
    private boolean running = true;
    private final ConsumerCallback callBack;
    /**
     * 日志对象
     */
    protected Logger logger = LoggerFactory.getLogger(ConsumerLoop.class);
    Clock clock = Clock.systemUTC();


    public ConsumerLoop(String topic, String groupId, ConsumerCallback callBack) {

        this.callBack = callBack;
        this.topic = topic;
        this.consumer = KafkaConsumerHelper.getConsumer(groupId, topic);
        new Thread(this).start();
    }

    /**
     * 关闭循环
     */
    public void shutdown() {
        if (consumer != null) {
            running = false;
            consumer.close();
        }
    }

    @Override
    public void run() {
        consumer.subscribe(Lists.newArrayList(topic));

        while (running) {
            try {
                ConsumerRecords<String, String> records = consumer.poll(100);

                //在处理之前就提交offset,如果业务代码处理时间非常长,就会导致消费者组之间的心跳失败,然后触发消费者组的reblance,处理完成之后就无法提交offse
                //从而导致重复消费数据。
                if (!records.isEmpty()) {
                    consumer.commitSync();
                    for (ConsumerRecord<String, String> record : records) {
                        long start = clock.millis();
                        try {
                            callBack.process(record);
                        } catch (Exception e) {

                            //不做处理
                            logger.error("consumer processing throw a exception,record value is:" + record.value(), e);
                        }
                        long end = clock.millis();
                        logger.debug("topic[topic:" + record.topic() + ",partition:" + record.partition() + "," +
                                "offset:" + record.offset() +
                                "] processing is complete,spent time is:" + (end - start) + "(ms)");
                    }
                }
            } catch (Throwable e) {
                logger.error("consumer thread error", e);
            }
        }
    }
}